/*
 * Copyright (c) 2020, 2025, Oracle and/or its affiliates.
 *
 * This program is free software; you can redistribute it and/or modify
 * it under the terms of the GNU General Public License, version 2.0,
 * as published by the Free Software Foundation.
 *
 * This program is designed to work with certain software (including
 * but not limited to OpenSSL) that is licensed under separate terms,
 * as designated in a particular file or component or in included license
 * documentation.  The authors of MySQL hereby grant you an additional
 * permission to link the program and your derivative works with the
 * separately licensed software that they have either included with
 * the program or referenced in the documentation.
 *
 * This program is distributed in the hope that it will be useful,  but
 * WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See
 * the GNU General Public License, version 2.0, for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with this program; if not, write to the Free Software Foundation, Inc.,
 * 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA
 */

#include "modules/util/dump/dump_writer.h"

#include <exception>
#include <stdexcept>
#include <utility>

#include "mysqlshdk/libs/utils/logger.h"
#include "mysqlshdk/libs/utils/utils_net.h"

#include "modules/util/dump/dump_errors.h"

namespace mysqlsh {
namespace dump {

using mysqlshdk::storage::IFile;
using mysqlshdk::storage::Mode;

namespace {

constexpr uint64_t k_write_idx_every = 1024 * 1024;  // bytes

}  // namespace

Dump_write_result &Dump_write_result::operator+=(const Dump_write_result &rhs) {
  m_data_bytes += rhs.m_data_bytes;
  m_bytes_written += rhs.m_bytes_written;
  m_rows_written += rhs.m_rows_written;

  return *this;
}

Dump_writer::Buffer::Buffer()
    : m_data(std::make_unique<char[]>(m_capacity)), m_ptr(m_data.get()) {}

void Dump_writer::Buffer::append_fixed(const std::string &s) noexcept {
  const auto length = s.length();

  assert(m_fixed_length_remaining >= length);

  append(s.c_str(), length);

  m_fixed_length_remaining -= length;
}

void Dump_writer::Buffer::clear() noexcept {
  m_ptr = m_data.get();
  m_length = 0;
  m_fixed_length_remaining = m_fixed_length;
}

void Dump_writer::Buffer::set_fixed_length(std::size_t fixed_length) {
  will_write(fixed_length);

  m_fixed_length_remaining = m_fixed_length = fixed_length;
}

void Dump_writer::Buffer::will_write(std::size_t bytes) {
  const auto requested_capacity = m_length + m_fixed_length_remaining + bytes;

  if (requested_capacity > m_capacity) {
    resize(requested_capacity);
  }
}

void Dump_writer::Buffer::resize(std::size_t requested_capacity) {
  std::size_t new_capacity = m_capacity;

  while (new_capacity < requested_capacity) {
    new_capacity <<= 1;
  }

  if (new_capacity != m_capacity) {
    auto new_data = std::make_unique<char[]>(new_capacity);
    memcpy(new_data.get(), m_data.get(), m_length);

    m_capacity = new_capacity;
    m_data = std::move(new_data);
    m_ptr = m_data.get() + m_length;
  }
}

void Dump_writer::Buffer::write_base64_data(const char *data,
                                            std::size_t length) {
  if (!length) {
    return;
  }

  // this function is meant to be used with base64 encoded data generated by
  // MySQL, which has lines exactly 76 chars long, except for the last one.
  // rfc2045 says at most 76 chars, so this will NOT work with arbitrary base64
  // data.
  static constexpr auto k_line_length = 76;

  will_write(length);

  // the base64 decoder doesn't really require newlines so just strip them
  auto p = data;
  const auto end = data + length - k_line_length;

  while (p < end) {
    append(p, k_line_length);
    p += k_line_length + 1;
  }

  // append the leftover
  append(p, data + length - p);
}

Dump_writer::Dump_writer() : m_buffer(std::make_unique<Buffer>()) {}

Dump_writer::~Dump_writer() {
  try {
    close();
  } catch (const std::runtime_error &error) {
    log_error("During destruction of Dump_writer: %s", error.what());
  }
}

void Dump_writer::set_output_file(mysqlshdk::storage::IFile *output) {
  m_output = output;
  m_compressed = dynamic_cast<mysqlshdk::storage::Compressed_file *>(m_output);
}

void Dump_writer::set_index_file(
    std::unique_ptr<mysqlshdk::storage::IFile> index) {
  m_index = std::move(index);
}

void Dump_writer::open() {
  if (m_index && !m_index->is_open()) {
    m_index->open(Mode::WRITE);
  }
}

void Dump_writer::close() {
  if (m_index && m_index->is_open()) {
    // write the total uncompressed size
    write_index();
    m_index->close();
  }
}

void Dump_writer::initialize_writing(
    const std::vector<mysqlshdk::db::Column> &metadata,
    const std::vector<Encoding_type> &pre_encoded_columns) {
  read_metadata(metadata, pre_encoded_columns);
}

Dump_write_result Dump_writer::write_preamble() {
  buffer()->clear();
  store_preamble();
  return write_buffer("preamble");
}

Dump_write_result Dump_writer::write_row(const mysqlshdk::db::IRow *row) {
  buffer()->clear();
  store_row(row);
  auto result = write_buffer("row", true);

  m_bytes_written += result.data_bytes();
  m_bytes_written_per_idx += result.data_bytes();

  if (m_index && m_bytes_written_per_idx >= k_write_idx_every) {
    write_index();
    // make sure offsets are written when close to the k_write_idx_every
    m_bytes_written_per_idx %= k_write_idx_every;
  }

  return result;
}

Dump_write_result Dump_writer::write_postamble() {
  buffer()->clear();
  store_postamble();
  return write_buffer("postamble");
}

Dump_write_result Dump_writer::write_buffer(const char *context,
                                            bool row) const {
  assert(m_output);

  Dump_write_result result;

  result.write_data(buffer()->length());

  if (row) {
    result.write_row();
  }

  if (result.data_bytes() > 0) {
    const auto bytes_written =
        m_output->write(buffer()->data(), result.data_bytes());

    if (bytes_written < 0) {
      THROW_ERROR(SHERR_DUMP_DW_WRITE_FAILED, context,
                  m_output->full_path().masked().c_str());
    }

    result.write_bytes(m_compressed ? m_compressed->latest_io_size()
                                    : bytes_written);
  }

  return result;
}

void Dump_writer::write_index() {
  assert(m_index);

  // the idx file contains offsets to the data stream, not to binary one
  const auto offset = mysqlshdk::utils::host_to_network(m_bytes_written);
  m_index->write(&offset, sizeof(uint64_t));
}

}  // namespace dump
}  // namespace mysqlsh
